package cn.net.wanji.controller;


import cn.net.wanji.cache.CacheManager;
import cn.net.wanji.cache.IdentifyTypeCacheManager;
import cn.net.wanji.config.ApplicationConfig;
import cn.net.wanji.config.websocket.WebSocket;
import cn.net.wanji.utils.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.utils.Utils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

import static cn.net.wanji.config.CommonParamConfig.NULL_LICENSE;
import static cn.net.wanji.service.iml.MonitorServiceImpl.eventHashMap;

/**
 * @author xinyingjie
 */
@RestController
@Slf4j
public class KafkaController {

    @Autowired
    private WebSocket webSocketServer;

    private final CacheManager cacheManager;

    private final IdentifyTypeCacheManager identifyTypeCacheManager;

    public KafkaController(CacheManager cacheManager, IdentifyTypeCacheManager identifyTypeCacheManager) {
        this.cacheManager = cacheManager;
        this.identifyTypeCacheManager = identifyTypeCacheManager;
    }

    @KafkaListener(id = "#{T(java.util.UUID).randomUUID().toString()}", topics = "${kafka.receive-topic}")
    public void consumerAlarm(ConsumerRecord<?, ?> record) throws InterruptedException {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        try {
            if (kafkaMessage.isPresent()) {
                Object message = kafkaMessage.get();
                if (!Utils.isBlank((String) message)) {
                    String sendDataJson = String.valueOf(message);
                    JSONObject jsonObject = JSON.parseObject(String.valueOf(sendDataJson));
                    if (jsonObject.containsKey("objs")) {
                        JSONArray jsonNew = new JSONArray();
                        JSONArray jsonArray = jsonObject.getJSONArray("objs");
                        for (Object o : jsonArray) {
                            JSONObject json = JSON.parseObject(String.valueOf(o));
                            String id = "1111111";
                            String carId = json.getString("carId");
                            String licenseCode = json.getString("licenseCode");
                            String allTypeCode = json.getString("allTypeCode");
                            List<String> carTypes = identifyTypeCacheManager.get(carId, String.class);
                            if (carTypes != null) {
                                // 获取当前时间
                                LocalDateTime currentTime = LocalDateTime.now();
                                // 定义日期时间格式
                                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
                                // 格式化当前时间
                                String formattedTime = currentTime.format(formatter);
                                /*log.info("获取到的当前时间是：{},当前车辆id是：{}, 车牌是：{}透传的车辆类型: {},车辆类型是：{}",
                                        formattedTime, carId, licenseCode, allTypeCode, carTypes.get(0));
*/
                                json.put("typeCode", carTypes.get(0));
                                json.put("identifyType", carTypes.get(0));
                            }
                            List<String> licenseList;
                            if (StringUtils.isEmpty(licenseCode)) {
                                licenseList = cacheManager.get(licenseCode, String.class);
                            } else {
                                licenseList = cacheManager.get(licenseCode, String.class);
                            }
                            if (licenseList != null) {
                                id = licenseList.get(0);
                            }
                            json.put("eventType", eventHashMap.getOrDefault(id, "0"));
                            jsonNew.add(json);
                        }
                        jsonObject.put("objs", jsonNew);
                    }
                    String sendData = jsonObject.toJSONString();
                    webSocketServer.sendAllMessage(sendData);
                }

            }
        } catch (Exception e) {
            log.error("kafka数据转发异常：{}", e.getMessage());
        }

    }
}
